package com.zhang.flink.example;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;


/**
 * @title: 实时热门商品 纯Flink SQL实现方式
 * @author: zhang
 * @date: 2022/2/14 19:18
 */
public class ItemCountTopNOnFlinkSQL {
    public static void main(String[] args) throws Exception {
        // todo 获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);


        //todo 读取文件系统数据
        tableEnv.executeSql("CREATE TABLE item ( " +
                "  `userId` STRING, " +
                "  `itemId` STRING, " +
                "  `categoryId` STRING, " +
                "  `type` STRING, " +
                "  `ts` BIGINT ," +
                "  `rt` as TO_TIMESTAMP(FROM_UNIXTIME(ts))," +
                "   WATERMARK FOR rt AS rt - INTERVAL '0' SECOND ) with ( " +
                " 'connector' ='filesystem', " +
                " 'path' = 'file:///Users/apple/project-git/flink-example/src/main/resources/UserBehavior.csv' , " +
                " 'format' = 'csv'" + ")");

        // todo 分组开窗
        Table itemCountWindow = tableEnv.sqlQuery(
                "select " +
                        " itemId, " +
                        " count(itemId) as cnt," +
                        " HOP_START(rt,INTERVAL '5' MINUTES,INTERVAL '1' HOURS) as windowStart," +
                        " HOP_END(rt,INTERVAL '5' MINUTES,INTERVAL '1' HOURS) as windowEnd" +
                        " from " +
                        "  item " +
                        " group by itemId,HOP(rt,INTERVAL '5' MINUTES,INTERVAL '1' HOURS)");

        //todo 排序
        Table rkTable = tableEnv.sqlQuery(
                "select " +
                        " itemId," +
                        " cnt," +
                        " windowEnd," +
                        " row_number() over(partition by windowEnd order by cnt ) as rk" +
                        " from " + itemCountWindow
        );


       // todo  求TopN
        Table result = tableEnv.sqlQuery(
                "select" +
                        " * " +
                        "from " +
                        "" + rkTable + " " +
                        "where rk <= 3");

        //todo 打印表信息
        result.execute().print();
    }
}
